package com.webchat.common.service.messagequeue.consumer;

import com.webchat.common.enums.LUAScriptConstants;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.service.RedisService;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.util.JsonUtil;
import com.webchat.domain.dto.queue.BaseDelayNormalQueueDTO;
import com.webchat.domain.dto.queue.BaseDelayQueueDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.script.RedisScript;

import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 抽象的redis队列实现方法
 *
 * @param <T>
 */
@Slf4j
public abstract class AbstractRedisDelayQueueConsumer<T extends BaseDelayQueueDTO> {

    @Autowired
    protected RedisService redisService;

    private volatile boolean isInit = false;

    private volatile boolean isRun = true;

    private ThreadPoolExecutor poolExecutor;

    public void init() {
        if (isInit) {
            return;
        }
        log.info("开始初始化消息队列，参数如下: \n" +
                        "queue name: {} \n" +
                        "queue timeout: {}\n" +
                        "queue retry times: {}\n" +
                        "queue back queue times: {}\n" +
                        "queue pool size: {}",
                getMessageQueue().getQueue(), getTimeout(), getRetryTimes(), getBackQueueTimes(), getPoolSize()
        );
        initThreadPool(getPoolSize());
        Long startSize = redisService.zSetSize(getQueueName());
        log.info("启动消息队列{}成功, 目前还有 {} 个元素没被消费", getQueueName(), startSize);
        isInit = true;
    }

    /**
     * 从消息队列redis读取数据的延迟
     *
     * @return
     */
    protected long getTimeout() {
        return 3000;
    }

    /**
     * 一个任务元素重试的次数
     *
     * @return 默认不重试
     */
    protected int getRetryTimes() {
        return 0;
    }

    /**
     * 一个任务最多被重返队列的次数
     *
     * @return
     */
    protected int getBackQueueTimes() {
        return 0;
    }

    /**
     * 获取最大执行任务数量
     *
     * @return
     */
    protected int getPoolSize() {
        return 1;
    }

    /**
     * 将字符串转化为对象
     *
     * @param s
     * @return
     */
    protected abstract T convert(String s);

    /**
     * 队列信息
     *
     * @return
     */
    protected abstract MessageQueueEnum getMessageQueue();

    /**
     * 执行下一步消费实际处理逻辑
     */
    protected abstract void doNextConsume(Set<String> messages);

    protected String getQueueName() {
        return this.getMessageQueue().getQueue();
    }

    /**
     * 延迟队列消费
     *
     * @return
     */
    private Set<String> consumeMessage() {
        long currentTime = System.currentTimeMillis();
        /**
         * 这里的实现方式在并发下无法保证读、删两个动作的原子性，有并发安全问题
         *
         * 比如：
         * ----------------------------
         * 线程A： 读（A）-> 删（A）-> 返回
         * 线程B： 读（B）-> 删（B）-> 返回
         * 实际执行可能如下：
         * 读（A）-> 读（B）（这里可能出现重复消费问题）-> 删（A）-> 删（B）-> 返回
         */
        // 按照score升序取消息任务
//        Set<String> messages = redisService.zrangeByScore(getQueueName(), Long.MIN_VALUE, currentTime);
//        if (CollectionUtils.isNotEmpty(messages)) {
//            // 消费到了数据，需要删除队列中数据，否则会重复消费
//            redisService.zremoveRangeByScore(getQueueName(), Long.MIN_VALUE, currentTime);
//        }
        /**
         * LUA 脚本实现，保证并发消息安全性
         */
        String execRes = redisService.executeScript(RedisScript.of(LUAScriptConstants.REDIS_DELAY_QUEUE_MESSAGE_CONSUME, String.class),
                List.of(getQueueName()),
                String.valueOf(Long.MIN_VALUE), String.valueOf(currentTime));
        if (StringUtils.isBlank(execRes)) {
            return Collections.emptySet();
        }
        return JsonUtil.fromJson(execRes, LinkedHashSet.class);
    }

    /**
     * 负责调度的核心方法
     * 通过循环读取redis的阻塞队列，获取元素，并且转化为用户可直接操作的类型
     * 如果处理失败采用两种容错方式:
     * 1. 重试，可以通过设置retryTime来设置重试次数
     * 2. 重新归队，只有在重试全部失败后才回执行该逻辑，可以设置backQueueTimes设置归队的最大次数
     * 如果两种方式都失败，则进入用户自定义的容错方式中
     */
    public void schedule() {
        init();
        try {
            final Semaphore semaphore = new Semaphore(getPoolSize());
            while (isRun) {
                Thread.sleep(1000);
                // 判断是否目前达到了使用上限
                log.info("QueueName:{} 信号量剩余:{}", getQueueName(), semaphore.availablePermits());
                semaphore.acquire();
                // 消费消息
                final Set<String> messages = this.consumeMessage();
                if (CollectionUtils.isEmpty(messages)) {
                    log.info("QueueName:{} 目前没有元素", getQueueName());
                    semaphore.release();
                    continue;
                }
                poolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            doNextConsume(messages);
                        } finally {
                            // 任务执行完释放信号量
                            semaphore.release();
                        }
                    }
                });
            }
        } catch (final Exception outerEx) {
            log.error("启动消息队列失败, 原因 ", outerEx);
        }
    }

    /**
     * 初始化数据库连接池
     *
     * @param poolSize
     */
    private void initThreadPool(int poolSize) {
        if (poolExecutor == null) {
            synchronized (this) {
                if (poolExecutor == null) {
                    poolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolSize);
                }
            }
        }
    }
}
